package com.study.flink.java.day04_async.sink.mysql;

import com.study.flink.java.day04_async.ActivityBean;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class MysqlSink extends RichSinkFunction<ActivityBean> {

    private transient Connection conn = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        conn = DriverManager.getConnection("jdbc:mysql://node02:3306/bigdata?characterEncoding=UTF-8", "linys", "123456");
    }

    @Override
    public void close() throws Exception {
        super.close();
        conn.close();
    }

    @Override
    public void invoke(ActivityBean bean, Context context) throws Exception {
        PreparedStatement pstm = conn.prepareStatement("insert into t_activity_count (aid, event_type, counts) values (?,?,?) on duplicate key update counts = ?");
        pstm.setString(1, bean.aid);
        pstm.setInt(2, bean.eventType);
        pstm.setInt(3, bean.count);
        pstm.setInt(4, bean.count);
        pstm.executeUpdate();
        pstm.close();
    }
}
